/* This class handles an incomming connection, gets the incomming messages and
 * puts them into inputbuffer.
 * It also adds the socket to the socket hashtable,if it doesn't exists.
 */

package org.kth.dks.dks_comm;

import java.io.DataInputStream;
import java.io.IOException;
import java.net.Socket;

import org.apache.log4j.Logger;
import org.kth.dks.dks_marshal.BootstrapMsg;
import org.kth.dks.dks_marshal.DKSMarshal;
import org.kth.dks.dks_marshal.DKSMessage;

public class ConnHandlerIn extends Thread {
    private Logger log = Logger.getLogger(ConnHandlerIn.class);
	private Socket socket                      = null;
    private DKSNetAddress destRef              = null;
    private ConnHandlerOut connOut             = null; // INVARIANT: must be set by setConnOut() always!
    private final DKSMarshal marshal;
    private ConnectionHandler connHandler      = null;

    // Fixed such that a connectionhandler in always has a ConnHandler out when it is started
    public ConnHandlerIn(ConnectionHandler ch, Socket newSocket, DKSNetAddress dest, ConnHandlerOut out) {
      connHandler = ch;
      marshal     = connHandler.getDKSMarshal();
      this.socket = newSocket;
      this.destRef = dest;
      this.connOut = out; 
      if (dest == null){
    	  log.debug( "Creating a new received - accepting");
      }
      else{
    	  log.debug("Creating a new received - connecting:" + dest.toString());
      }
      this.setName(ConnHandlerIn.class.getName() + "(" + dest + ")");
      this.start();
    }

    public void setConnOut(ConnHandlerOut c) {
      connOut=c;
    }

  void end() {
    try {
      socket.close();
    }
    catch (IOException ex) {
      ex.printStackTrace();
    }
  }

  public void run() {
    DataInputStream in = null;
    try {
      in = new DataInputStream(socket.getInputStream());
    } catch (IOException ioexp) {
      ioexp.printStackTrace();
    } //try
    
    if(connOut == null){
    	log.fatal("ConnHandlerIn: Major fault connOut == null"); 
    	System.exit(0); 
    }
    
    int msgID = 0;

    /*
    type=readInt;
    if (CONTENT) 
    	transType = readbytE;
        id = readInt;
        ReadBlock;
    else if (ACK)
    	id = readInt;
    else if (PRESENT)
    	READBLOCK;
    */
    
    while ( socket.isConnected() ) {
      try {
        // Reads one message (format: <type>)
        byte msgTypeByte = in.readByte();
        ConnMessageTypes msgType = ConnMessageTypes.valueOf(msgTypeByte);
        log.debug("MessageReceived " + msgType);
        if (msgType==ConnMessageTypes.CONTENTS_MSG) {
          //A message for higher levels <size><msgid><body>)
          byte transType = in.readByte();
          msgID = in.readInt();
          byte [] buffer = ConnMessageMiscs.m_readDataBlock(in);

          connOut.sendAck(msgID);
          if (false==marshal.unmarshalDispatch(transType, buffer, destRef))
        	  log.warn( "Msg not accepted, CommunicationBuffer overflow");
          connHandler.statAddMsgsReceived(1);
          connHandler.statAddBytesReceived(buffer.length);
        } else if ( msgType==ConnMessageTypes.ACK_MSG ) {
          int msgId = in.readInt();
          connOut.ackReceived(msgId);
          connHandler.statAddMsgsDelivered(1);
        } if ( msgType==ConnMessageTypes.PRESENT_MSG ) {
          byte buffer [] = ConnMessageMiscs.m_readDataBlock(in);
          BootstrapMsg m = (BootstrapMsg) DKSMessage.unmarshal(buffer);
          destRef = m.getNetAddress();
        }
      }
      catch (IOException ioexp) {
        connHandler.removeConnection(destRef);
        connOut.end();
        log.info("Socket "+destRef+" closed inside ConnHandlerIn()\n");
        break;
      }
      catch (Exception exp) {
        connHandler.removeConnection(destRef);
        log.info("Socket "+destRef+" closed inside ConnHandlerIn()\n");
        break;
      }
    }
    log.debug("Closing ConnHandlerIn to node "+destRef);

  }

  public DKSNetAddress getDKSNetAddress() {
    return destRef;
  }

}
